/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal/allocator.h"
#include "dal_fastdds/utils/EntityMappingManager.hpp"
#include "dal_fastdds/utils/log.hpp"
#include "dal_fastdds/listener_impl/DataReaderListenerImplDbhExtend.hpp"
#include "fastdds/dds/core/StackAllocatedSequence.hpp"
#include <fastdds/dds/subscriber/DataReader.hpp>

DataReaderListenerImplDbhExtend::DataReaderListenerImplDbhExtend(
    const message_type_support_callbacks_t* ts_callbacks_,
    dal_dmabufheap::BufferDescriptorRingBuffer* ringbuffer)
    : dbh_recv_notify_msg_ts_callbacks_(ts_callbacks_), ringbuffer_(ringbuffer)
{
    const char* env_var_value = std::getenv("DAL_DBH_DEBUG");
    if (env_var_value != nullptr && std::atoi(env_var_value) == 1) {
        is_debug_print_enabled_ = true;
    }
}

DataReaderListenerImplDbhExtend::~DataReaderListenerImplDbhExtend() {}

void DataReaderListenerImplDbhExtend::on_data_available(DataReader* reader)
{
    SampleInfoSeq info_seq{1};
    dal_fastrtps_shared_cpp::SerializedData sdata;
    sdata.is_cdr_buffer = false;
    sdata.data = &dbh_recv_notify_msg_;
    sdata.impl = dbh_recv_notify_msg_ts_callbacks_;
    eprosima::fastdds::dds::StackAllocatedSequence<void*, 1> data_values;
    const_cast<void**>(data_values.buffer())[0] = &sdata;

    ReturnCode_t ret;
    while (true) {
        ret = reader->take(data_values, info_seq, 1);
        if (ret == ReturnCode_t::RETCODE_NO_DATA) {
            break;
        }

        if (ret != ReturnCode_t::RETCODE_OK) {
            LOG_ERROR_WITH_LOCATION()
                << "Error reading data from datareader ret = " << ret() << std::endl;
            return;
        }

        if (is_debug_print_enabled_) {
            std::cout << "[recv notify reader <------]  topic = "
                      << reader->get_topicdescription()->get_name() << ", UUID = ";
            for (const auto& byte : dbh_recv_notify_msg_.uuid) {
                std::cout << std::hex << +byte << " ";
            }
            std::cout << std::dec;
        }

        if (ringbuffer_) {
            std::lock_guard<std::mutex> lock(ringbuffer_->ringbuffer_mutex_);
            ringbuffer_->DecreaseByUUID(dbh_recv_notify_msg_.uuid.data());

            if (is_debug_print_enabled_) {
                std::cout << ", ringbuffer index = "
                          << ringbuffer_->GetIndexByUUID(dbh_recv_notify_msg_.uuid.data())
                          << ", ref_cnt = "
                          << ringbuffer_->GetRefCntByUUID(dbh_recv_notify_msg_.uuid.data());
            }
        }

        if (is_debug_print_enabled_) {
            std::cout << std::endl;
        }

        continue;
    }
}
